package cn.gwm.flink.streaming.task;

import cn.gwm.flink.streaming.beans.BeanSource;
import cn.gwm.flink.streaming.constant.BaseFields;
import cn.gwm.flink.streaming.constant.FaultToleranceConstant;
import cn.gwm.flink.streaming.dwm.charge.util.ModeCodeUtil;
import cn.gwm.flink.streaming.sink.kafka.FlinkKafkaUtil;
import cn.gwm.utils.ConfigLoader;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Map;

/**
 * @Author: liangguang
 * @Date: 2022/12/26
 * @Description: 充电多维模型
 */
public class MultidimensionalModelChargeTask {

    public static void main(String[] args) throws Exception {
        ConfigLoader.init(args);
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        BaseTask.envSet(env, ConfigLoader.get(FaultToleranceConstant.FLINK_JOB_NAME));


        // 1.充电主题重跑，只需要 充电明细
        final FlinkKafkaConsumer<String> kafkaConsumer = FlinkKafkaUtil.getConsumerGroupOffsets(
                "consumer-dwm-charge-multidimensional",
                ConfigLoader.get("kafka.topic.charge.detail.ev", "bms_charge_dwm_detail")
        );
        DataStream<JSONObject> sourceDataStream = env.addSource(kafkaConsumer).map(JSONUtil::parseObj);
        SingleOutputStreamOperator<JSONObject> detail = sourceDataStream
                .filter(item -> "1".equals(item.getStr("OBC_ConnectSts_EV")))
                .map(new supplementModeCode())
                .keyBy(sd -> sd.getStr(BaseFields.vin))
                .map(new ExtractCharge())
                .filter(item -> item != null)
                .map(item -> BeanSource.transJson(item, BeanSource.SourceEnum.evChargeMultidimensional));

        detail.addSink(FlinkKafkaUtil.toKafka(ConfigLoader.get("kafka.multidimensional.topic.dwm")));

        // hive存储
        detail.addSink(BaseTask.getFileSink(ConfigLoader.get("hive.table.name.dwm.charge.multidimensional"),
                ".txt",
                "dwm/" + ConfigLoader.get("hive.table.name.dwm.charge.multidimensional"),
                BeanSource.SourceEnum.evChargeMultidimensional));

        env.execute();
    }


    public static class supplementModeCode extends RichMapFunction<JSONObject, JSONObject> {

        @Override
        public JSONObject map(JSONObject item) throws Exception {
            if (BaseFields.defaultVal.equals(item.getStr("model_code"))) {
                String vin = item.getStr(BaseFields.vin);
                // 如果是默认值，则需要补充
                Map<String, String> map = ModeCodeUtil.batteryData(vin);
                item.set("model_code", map.get("modelCode"));
            }
            return item;
        }
    }


    /**
     * 多维模型 数据抽取
     */
    public static class ExtractCharge extends RichMapFunction<JSONObject, JSONObject> {
        private ValueState<Long> startTidState;
        private MapState<Double, String> bmsSocState;

        @Override
        public void open(Configuration parameters) throws Exception {
            startTidState = getRuntimeContext().getState(new ValueStateDescriptor<>("startTidState", Long.class));
            bmsSocState = getRuntimeContext().getMapState(new MapStateDescriptor<Double, String>("bmsSocState", Double.class, String.class));
        }

        @Override
        public JSONObject map(JSONObject entries) throws Exception {
            double bms_soc = entries.getDouble("BMS_SOC");
            Long startTid = entries.getLong("startTid");
            double remainder = bms_soc % 0.5d;
            if (startTidState.value() != null) {
                Long startTidVal = startTidState.value();
                if (startTidVal.equals(startTid)) {
                    // 同一次物理充电
                    if (0.0d == remainder) {
                        // 整除了
                        boolean contains = bmsSocState.contains(bms_soc);
                        if (!contains) {
                            bmsSocState.put(bms_soc, BaseFields.defaultVal);
                            return entries;
                        }
                    }
                } else {
                    // 非同一次 物理充电
                    startTidState.update(startTid);
                    bmsSocState.clear();
                    if (0.0d == remainder) {
                        bmsSocState.put(bms_soc, BaseFields.defaultVal);
                        return entries;
                    }
                }

            } else {
                startTidState.update(startTid);
                bmsSocState.clear();
                if (0.0d == remainder) {
                    bmsSocState.put(bms_soc, BaseFields.defaultVal);
                    return entries;
                }
            }
            return null;
        }
    }

}
